package com.shujia.flink.tf;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class Demo2FlatMap {
    public static void main(String[] args) throws Exception {
        //1、创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2、读取数据
        DataStream<String> linesDS = env.readTextFile("data/words");

        //3、一行转换成多行
        DataStream<String> wordsDS = linesDS.flatMap(new FlatMapFunction<String, String>() {

            /**
             * flatMap方法每一条数据执行一次，可以返回多条数据
             * @param line：一行数据
             * @param collector：用于将数据发送到下游
             */
            @Override
            public void flatMap(String line, Collector<String> collector) throws Exception {
                //切分字符串
                String[] split = line.split(",");

                //循环将数据发送到下游
                for (String word : split) {
                    //将数据发送到下游
                    collector.collect(word);
                }

            }
        });


        //java lambda表达式
        //InvalidTypesException:如果出现错误，需要增加返回类型
        // TypeInformation.of(String.class) : 指定算子返回类型
        DataStream<String> wordsDS1 = linesDS.flatMap((line, collector) -> {
            String[] split = line.split(",");
            for (String word : split) {
                collector.collect(word);
            }
        }, TypeInformation.of(String.class));

        wordsDS1.print();
        env.execute();

    }
}
